package com.zhxy.util.message;

import java.io.Serializable;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledThreadPoolExecutor;

import org.apache.mina.transport.socket.nio.NioSocketAcceptor;
import org.apache.mina.transport.socket.nio.NioSocketConnector;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;

import com.zhxy.util.async.ResultHandler;
import com.zhxy.util.message.AsyncMessageProcessor;
import com.zhxy.util.message.DuplexMessagePipe;
import com.zhxy.util.message.DuplexMessageServiceListener;
import com.zhxy.util.message.MessageCodec;
import com.zhxy.util.message.MinaDuplexMessageService;
import com.zhxy.util.message.jsonpack.JsonpackCodec;

public class MinaDuplexMessageServiceTestCase {
	
	private static MinaDuplexMessageService server = new MinaDuplexMessageService();
	
	private static MinaDuplexMessageService client = new MinaDuplexMessageService();
	
	private static volatile DuplexMessagePipe clientPipe;
	
	private static MessageCodec codec;
	
	private static ScheduledThreadPoolExecutor executor;

	@BeforeClass
	public static void breforeClass() throws Exception {
		
		executor = new ScheduledThreadPoolExecutor(1);
		
		JsonpackCodec cc = new JsonpackCodec();
		cc.registerType("msg", MyMessage.class);
		codec = cc;
		
		server.setCodec(codec);
		server.setScheduledExecutor(executor);
		server.addDuplexMessageServiceListener(new DuplexMessageServiceListener() {
			public void disconnected(DuplexMessagePipe pipe) {}
			public void connected(DuplexMessagePipe pipe) {
				
			}
		});
		
		client.setCodec(codec);
		client.setScheduledExecutor(executor);
		client.addMessageProcessor(new Processor());
	}
	
	@AfterClass
	public static void afterClass() throws Exception {
		client.close();
		server.close();
		executor.shutdownNow();
	}
	
	
	@Before
	public void brefore() throws Exception {
		
	}
	
	@After
	public void after() throws Exception {

	}
	
	@Test
	public void testConnect() throws Exception {
		
		final Object lock = new Object();
		DuplexMessageServiceListener listener = new DuplexMessageServiceListener() {
			public void disconnected(DuplexMessagePipe pipe) {}
			public void connected(DuplexMessagePipe pipe) {
				synchronized (lock) {
					clientPipe = pipe;
					lock.notify();
				}
			}
		};
		
		NioSocketAcceptor acceptor = new NioSocketAcceptor();
		server.bindIoService(acceptor);
		acceptor.bind(new InetSocketAddress(15678));
		
		NioSocketConnector connector = new NioSocketConnector();
		client.addDuplexMessageServiceListener(listener);
		client.bindIoService(connector);
		connector.connect(new InetSocketAddress("127.0.0.1", 15678));
		synchronized (lock) {
			lock.wait();
		}
		client.removeDuplexMessageServiceListener(listener);
	}
	
	@Test
	public void testSend() throws Exception {
		
		long loop = 1000;
		RecordProcessor processor = new RecordProcessor();
		server.addMessageProcessor(processor);
		
		for(int i = 0; i < loop; i++) {
			MyMessage message = new MyMessage();
			message.name = "command " + i;
			clientPipe.send(message);
		}
		
		for (int i = 0; i < 10; i++) {
			if (processor.recvCommandQueue.size() >= loop) {
				break;
			}
			Thread.sleep(100);
		}
		Assert.assertEquals(loop, processor.recvCommandQueue.size());
	}
	
	@Test
	public void testRequest() throws Exception {
		
		long loop = 1000;
		RecordProcessor processor = new RecordProcessor(){
			@Override
			protected Object getResponse(Object message) {
				MyMessage request = (MyMessage)message;
				MyMessage response = new MyMessage();
				response.name = request.name;
				return response;
			}
		};
		server.addMessageProcessor(processor);
		
		
		class Handler implements ResultHandler<Object> {
			private String name;
			public Handler(String name) {
				this.name = name;
			}
			public void result(Object result) {
				MyMessage msg = (MyMessage)result;
				Assert.assertEquals(msg.name, name);
			}
			public void exception(Throwable e) {
			}
		}
		
		for(int i = 0; i < loop; i++) {
			MyMessage message = new MyMessage();
			message.name = "command " + i;
			clientPipe.request(message, new Handler(message.name));
		}
		
		for (int i = 0; i < 10; i++) {
			if (processor.recvRequestQueue.size() >= loop) {
				break;
			}
			Thread.sleep(100);
		}
		Assert.assertEquals(loop, processor.recvRequestQueue.size());
	}
	
	
	@Test
	public void test4() throws Exception {
		
		// 需要测试一服务器对多客户端的(请求->回应)方式的消息
		
		MinaDuplexMessageService server = new MinaDuplexMessageService();
		server.setCodec(codec);
		server.setScheduledExecutor(executor);
		RecordProcessor serverRecord = new RecordProcessor();
		server.addMessageProcessor(serverRecord);
		
		NioSocketAcceptor acceptor = new NioSocketAcceptor();
		server.bindIoService(acceptor);
		acceptor.bind(new InetSocketAddress(27440));
		
		//
		final LinkedBlockingQueue<RecordProcessor> records = new LinkedBlockingQueue<RecordProcessor>();
		
		int nodeCount = 5;
		ArrayList<Thread> threads = new ArrayList<Thread>();
		for (int i = 0; i < nodeCount; i++) {
			Thread t = new Thread(new Runnable() {
				public void run() {
					
					RecordProcessor clientRecord = new RecordProcessor();
					
					MinaDuplexMessageService client = new MinaDuplexMessageService();
					client.setCodec(codec);
					client.setScheduledExecutor(executor);
					client.addMessageProcessor(clientRecord);
					NioSocketConnector connector = new NioSocketConnector();
					client.bindIoService(connector);
					connector.connect(new InetSocketAddress("127.0.0.1", 27440));
					clientRecord.setService(client);
					try {
						Thread.sleep(500);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
					
					long sentRequestCount = 0;
					Random r = new Random();
					for (int i = 0; i < 1000; i++) {
						if (r.nextBoolean()) {
							clientRecord.sendRequest("request " + i);
							sentRequestCount++;
						} else {
							clientRecord.sendCommand("command " + i);
						}
					}
					records.add(clientRecord);
					
					try {
						for (int i = 0; i < 100; i++) {
							if (clientRecord.recvResponseQueue.size() >= sentRequestCount) {
								break;
							}
							Thread.sleep(100);
						}
						Assert.assertEquals(sentRequestCount, clientRecord.recvResponseQueue.size());
						client.close();
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
			});
			t.start();
			threads.add(t);
		}
		for (Thread t : threads) {
			t.join();
		}
		
		//Thread.sleep(7000);
		Assert.assertEquals(nodeCount, records.size());
		
		long totalSendCommand = 0;
		for (RecordProcessor node : records) {
			totalSendCommand += node.sendCommandQueue.size();
			Assert.assertEquals(node.sendRequestQueue.size(), node.recvResponseQueue.size());
			for (int i = 0; i < node.sendRequestQueue.size(); i++) {
				String request = (String)node.sendRequestQueue.take();
				String response = (String)node.recvResponseQueue.take();
				Assert.assertEquals("reply for " + request, response);
			}
		}
		Assert.assertEquals(totalSendCommand, serverRecord.recvCommandQueue.size());
	}
	
	public static class MyMessage implements Serializable {
		private static final long serialVersionUID = 1L;
		private String name;
		public String getName() {
			return name;
		}
		public void setName(String name) {
			this.name = name;
		}
	}

	public static class Processor implements AsyncMessageProcessor<DuplexMessagePipe, Object> {

		@Override
		public void process(DuplexMessagePipe context, Object message,
				ResultHandler<Object> handler) {
			
			MyMessage msg = (MyMessage)message;
			
			if (handler != null) {
				System.out.println("reply Message:" + msg.name);
				MyMessage result = new MyMessage();
				result.name = "response for " + msg.name;
				handler.result(result);
			} else {
				System.out.println("process Message:" + msg.name);
			}
		}
		
	}
	
	
	public static class RecordProcessor implements AsyncMessageProcessor<DuplexMessagePipe, Object> {
		
		private MinaDuplexMessageService service;
		
		private BlockingQueue<Object> sendCommandQueue = new LinkedBlockingQueue<Object>();
		
		private BlockingQueue<Object> sendRequestQueue = new LinkedBlockingQueue<Object>();
		
		
		private BlockingQueue<Object> recvCommandQueue = new LinkedBlockingQueue<Object>();
		
		private BlockingQueue<Object> recvRequestQueue = new LinkedBlockingQueue<Object>();
		
		private BlockingQueue<Object> recvResponseQueue = new LinkedBlockingQueue<Object>();
		
		public MinaDuplexMessageService getService() {
			return service;
		}

		public void setService(MinaDuplexMessageService service) {
			this.service = service;
		}
		
		public BlockingQueue<Object> getSendCommandQueue() {
			return sendCommandQueue;
		}

		public BlockingQueue<Object> getSendRequestQueue() {
			return sendRequestQueue;
		}

		public BlockingQueue<Object> getRecvCommandQueue() {
			return recvCommandQueue;
		}

		public BlockingQueue<Object> getRecvRequestQueue() {
			return recvRequestQueue;
		}

		public BlockingQueue<Object> getRecvResponseQueue() {
			return recvResponseQueue;
		}

		public void sendCommand(Object message) {
			service.getPipes().get(0).send(message);
			sendCommandQueue.add(message);
		}
		
		public void sendRequest(Object message) {
			service.getPipes().get(0).request(message, new ResultHandler<Object>() {
				public void result(Object result) {
					recvResponseQueue.add(result);
				}
				public void exception(Throwable e) {
				}
			});
			sendRequestQueue.add(message);
		}

		@Override
		public void process(DuplexMessagePipe context, Object message,
				ResultHandler<Object> handler) {
			
			if (handler == null) {
				recvCommandQueue.add(message);
			} else {
				recvRequestQueue.add(message);
				handler.result(getResponse(message));
			}
			
		}
		
		protected Object getResponse(Object message) {
			String msg = (String)message;
			return "reply for " + msg;
		}
	}
}
